package org.apache.lucene.ocean.log;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.ReentrantLock;
import java.util.zip.CRC32;
import java.util.zip.CheckedInputStream;
import java.util.zip.CheckedOutputStream;

import org.apache.commons.io.output.ByteArrayOutputStream;
import org.apache.commons.io.output.CountingOutputStream;
import org.apache.lucene.ocean.LogDirectory;
import org.apache.lucene.ocean.util.Bytes;
import org.apache.lucene.ocean.util.OceanRandomAccessFile;
import org.apache.lucene.ocean.util.VDataInputStream;
import org.apache.lucene.ocean.util.VDataOutputStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Performs raw file access including serializing records to and from the
 * underlying file
 * 
 */
// TODO: add version and some header info to file
public class RawLogFile {
  final static Logger LOG = LoggerFactory.getLogger(RawLogFile.class);
  public static final byte DELETED = 101;
  public static final byte OK = 9;
  // public static final byte[] HEADER = new byte[] { 'O', 'c', 'N' };
  private long currentWritePosition = 0;
  private ReentrantLock writeLock = new ReentrantLock();
  private String file;
  private OceanRandomAccessFile output;
  private LogDirectory logDirectory;
  private byte[] crcBuffer = new byte[1024];

  public RawLogFile(String file, LogDirectory logDirectory) throws IOException {
    this.file = file;
    this.logDirectory = logDirectory;
    output = logDirectory.getOutput(file, false);
    // TODO: need better way to make sure we're starting at the end of the file
    currentWritePosition = logDirectory.fileLength(file);
    LOG.info("starting currentWritePosition: " + currentWritePosition);
  }

  public void resetWritePosition(List<RecordHeader> recordHeaders) {
    writeLock.lock();
    try {
      if (recordHeaders.size() > 0) {
        RecordHeader lastHeader = recordHeaders.get(recordHeaders.size() - 1);
        currentWritePosition = lastHeader.headerPosition + lastHeader.headersLength + lastHeader.docsLength + lastHeader.otherLength;
      } else {
        currentWritePosition = 0; // TODO: change when file header added
      }
    } finally {
      writeLock.unlock();
    }
  }

  public String getName() {
    return file;
  }

  public boolean delete() throws IOException {
    return logDirectory.deleteFile(file);
  }

  public OceanRandomAccessFile openInput() throws IOException {
    return logDirectory.openInput(file);
  }

  public long size() throws IOException {
    return logDirectory.fileLength(file);
  }

  public static interface StreamRecord {
    public StreamData getDocuments();

    public StreamData getOther();
  }

  /**
   * Throws an error if the crcs do not match and the read is at the end of the
   * stream.
   * 
   */
  public static class CRCInputStream extends CheckedInputStream {
    private long matchCrc;
    CRC32 crc = new CRC32();

    public CRCInputStream(InputStream input, long matchCrc) {
      super(input, new CRC32());
      this.matchCrc = matchCrc;
    }

    public long skip(long n) throws IOException {
      byte[] buf = new byte[512];
      long total = 0;
      while (total < n) {
        long len = n - total;
        len = read(buf, 0, len < buf.length ? (int) len : buf.length);
        if (len == -1) {
          return total;
        }
        total += len;
      }
      return total;
    }

    public int read() throws IOException {
      int b = in.read();
      if (b != -1) {
        crc.update(b);
      }
      if (b == -1)
        checkCrc();
      return b;
    }

    private void checkCrc() throws IOException {
      long inputCrc = crc.getValue();
      if (matchCrc != inputCrc) {
        throw new IOException("crc does not match");
      }
    }

    public int read(byte[] buf, int off, int len) throws IOException {
      len = in.read(buf, off, len);
      if (len != -1) {
        crc.update(buf, off, len);
      }
      if (len == -1)
        checkCrc();
      return len;
    }
  }

  public static class CRCException extends IOException {
    public CRCException(String msg) {
      super(msg);
    }
  }

  public static interface StreamData {
    public CRCInputStream getInputStream() throws IOException;

    public void verifyCrc() throws IOException, CRCException;

    public int getLength();
  }

  public static class FileStreamRecord implements StreamRecord {
    private RecordHeader recordHeader;
    private OceanRandomAccessFile input;

    public FileStreamRecord(RecordHeader recordHeader, OceanRandomAccessFile input) {
      this.recordHeader = recordHeader;
      this.input = input;
    }

    public FileStreamData getDocuments() {
      if (recordHeader.docsLength == 0)
        return null;
      long position = recordHeader.getDocsPosition();
      return new FileStreamData(recordHeader.docsLength, position, recordHeader.docsCrc32, input);
    }

    public FileStreamData getOther() {
      if (recordHeader.otherLength == 0)
        return null;
      long position = recordHeader.getOtherPosition();
      return new FileStreamData(recordHeader.otherLength, position, recordHeader.otherCrc32, input);
    }
  }

  public static class FileStreamData implements StreamData {
    private int length;
    private long position;
    private OceanRandomAccessFile input;
    private long crc32;

    public FileStreamData(int length, long position, long crc32, OceanRandomAccessFile input) {
      this.length = length;
      this.position = position;
      this.crc32 = crc32;
      this.input = input;
    }

    public long getCrc() {
      return crc32;
    }

    public void verifyCrc() throws IOException, CRCException {
      InputStream input = getInputStream();
      int numRead;
      byte[] buffer = new byte[1024];
      while ((numRead = input.read(buffer)) >= 0) {
      }
    }

    public CRCInputStream getInputStream() throws IOException {
      int bufferSize = 1024 * 8;
      if (bufferSize > length) {
        bufferSize = length;
      }
      return new CRCInputStream(new BufferedInputStream(input.getInputStream(position, length), bufferSize), crc32);
    }

    /**
     * public byte[] getBytes() throws IOException { input.seek(position);
     * byte[] bytes = new byte[length]; input.readFully(bytes); return bytes; }
     */
    public int getLength() {
      return length;
    }
  }

  public FileStreamRecord readRecordData(RecordHeader recordHeader, OceanRandomAccessFile input) {
    return new FileStreamRecord(recordHeader, input);
  }

  List<RecordHeader> loadRecordHeaders() throws IOException {
    OceanRandomAccessFile input = openInput();
    if (input.length() > 0) {
      LoadRecordHeaders loadRecordHeaders = new LoadRecordHeaders(input);
      return loadRecordHeaders.getRecordHeaders();
    } else {
      return new ArrayList<RecordHeader>();
    }
  }

  /**
   * public boolean delete(RecordHeader recordHeader) throws IOException { long
   * lastPosition = -1; writeLock.lock(); try { lastPosition =
   * output.getFilePointer(); output.seek(recordHeader.headerPosition); byte[]
   * header = new byte[3]; output.readFully(header, 0, header.length); if
   * (!Arrays.equals(HEADER, header)) { throw new IOException("no header"); }
   * byte first = (byte) output.readByte(); if (first == OK) { int
   * nextHeaderPosition = output.readInt(); long id = output.readLong(); assert
   * id == recordHeader.id; output.seek(recordHeader.headerPosition + 4);
   * output.writeByte(DELETED); output.getFD().sync(); return true; } else { //
   * already deleted return false; } } finally { if (lastPosition != -1)
   * output.seek(lastPosition); writeLock.unlock(); } }
   */
  /**
   * public class LoadRecordHeaders { private long currentReadPosition = 0;
   * private List<RecordHeader> recordHeaders = new ArrayList<RecordHeader>();
   * 
   * public LoadRecordHeaders(RandomAccessFile input) throws IOException { try {
   * while (true) { input.seek(currentReadPosition); RecordHeader recordHeader =
   * new RecordHeader(); byte[] header = new byte[3];
   * recordHeader.headerPosition = (int) input.getFilePointer();
   * input.readFully(header, 0, header.length); if (!Arrays.equals(HEADER,
   * header)) { // TODO: scan to next header if there is one LOG.error("header
   * incorrect: "+new String(header)+" at pos: "+currentReadPosition); while
   * (true) { recordHeader.headerPosition = (int) input.getFilePointer();
   * input.readFully(header, 0, header.length); if (Arrays.equals(HEADER,
   * header)) { LOG.error("found header at position:
   * "+recordHeader.headerPosition); break; } } // continue; } byte status =
   * input.readByte(); recordHeader.id = input.readLong(); if (status ==
   * DELETED) { // TODO: skip it if (LOG.isDebugEnabled()) LOG.debug("record
   * "+recordHeader.id+" status deleted"); } recordHeader.docsLength =
   * input.readInt(); recordHeader.otherLength = input.readInt();
   * recordHeader.docsPosition = input.getFilePointer(); if (status != DELETED)
   * recordHeaders.add(recordHeader); currentReadPosition =
   * input.getFilePointer() + recordHeader.docsLength +
   * recordHeader.otherLength; } } catch (EOFException eofException) { // at end
   * of file } finally { if (input != null) input.close(); } }
   * 
   * public List<RecordHeader> getRecordHeaders() { return recordHeaders; } }
   */

  public class LoadRecordHeaders {
    private long currentReadPosition = 0;
    private List<RecordHeader> recordHeaders = new ArrayList<RecordHeader>();
    private byte[] buffer = new byte[64];

    public LoadRecordHeaders(OceanRandomAccessFile input) throws IOException {
      try {
        while (true) {
          RecordHeader recordHeader = new RecordHeader();
          input.seek(currentReadPosition);
          recordHeader.headerPosition = (int) input.getFilePointer();
          int numRead = input.read(buffer);
          if (numRead == -1)
            break;
          ByteArrayInputStream byteInput = new ByteArrayInputStream(buffer);
          VDataInputStream firstVInput = new VDataInputStream(byteInput);
          long headerCrc32 = firstVInput.readVLong();
          int headersSize = firstVInput.readVInt();
          int initialSize = buffer.length - byteInput.available();
          // start reading from the buffer after the headercrc32 value
          // TODO: can reuse CRC32 object
          CheckedInputStream checkedInputStream = new CheckedInputStream(byteInput, new CRC32());
          VDataInputStream vInput = new VDataInputStream(checkedInputStream);
          byte status = vInput.readByte();
          recordHeader.headersLength = headersSize + initialSize;
          recordHeader.id = vInput.readVLong();
          recordHeader.docsLength = vInput.readVInt();
          recordHeader.docType = vInput.readVInt();
          recordHeader.docsCrc32 = vInput.readVLong();
          recordHeader.otherLength = vInput.readVInt();
          recordHeader.otherType = vInput.readVInt();
          recordHeader.otherCrc32 = vInput.readVLong();

          CRC32 crc32 = (CRC32) checkedInputStream.getChecksum();
          long crcValue = crc32.getValue();
          if (headerCrc32 != crcValue) {
            throw new IOException("header crc: " + headerCrc32 + " crc: " + crc32.getValue());
          }
          if (status == DELETED) {
            // TODO: skip it
            if (LOG.isDebugEnabled())
              LOG.debug("record " + recordHeader.id + " status deleted");
          }
          if (status != DELETED)
            recordHeaders.add(recordHeader);
          currentReadPosition = recordHeader.headerPosition + recordHeader.headersLength + recordHeader.docsLength
              + recordHeader.otherLength;
        }
      } catch (EOFException eofException) {
        // at end of file
      } finally {
        if (input != null)
          input.close();
      }
    }

    public List<RecordHeader> getRecordHeaders() {
      return recordHeaders;
    }
  }

  private static long getCrc32(byte[] bytes) {
    if (bytes == null)
      return -1;
    CRC32 crc32 = new CRC32();
    crc32.update(bytes);
    return crc32.getValue();
  }

  private long getCrc32(Bytes bytes) throws IOException {
    if (bytes == null)
      return -1;
    CRC32 crc32 = new CRC32();
    InputStream input = bytes.getInputStream();
    int numRead;
    while ((numRead = input.read(crcBuffer)) >= 0) {
      crc32.update(crcBuffer, 0, numRead);
    }
    return crc32.getValue();
  }

  public RecordHeader write(Long id, int docType, Bytes docBytes, int otherType, Bytes otherBytes) throws IOException {
    writeLock.lock();
    try {
      int documentsLength = 0;
      if (docBytes != null)
        documentsLength = docBytes.length();
      int otherLength = 0;
      if (otherBytes != null)
        otherLength = otherBytes.length();
      RecordHeader recordHeader = new RecordHeader();
      recordHeader.id = id;
      recordHeader.headerPosition = (int) currentWritePosition;
      recordHeader.docsLength = documentsLength;
      recordHeader.docType = docType;
      recordHeader.docsCrc32 = getCrc32(docBytes);
      recordHeader.otherLength = otherLength;
      recordHeader.otherType = otherType;
      recordHeader.otherCrc32 = getCrc32(otherBytes);

      output.seek(currentWritePosition);
      ByteArrayOutputStream byteOutput = new ByteArrayOutputStream(64); // can
      // be
      // reused
      CheckedOutputStream checkedOutputStream = new CheckedOutputStream(byteOutput, new CRC32());
      VDataOutputStream dataOutputStream = new VDataOutputStream(checkedOutputStream);
      dataOutputStream.writeByte(OK);
      // TODO: can do previous difference compression on these
      dataOutputStream.writeVLong(id);
      dataOutputStream.writeVInt(recordHeader.docsLength);
      dataOutputStream.writeVInt(recordHeader.docType);
      dataOutputStream.writeVLong(recordHeader.docsCrc32);
      dataOutputStream.writeVInt(recordHeader.otherLength);
      dataOutputStream.writeVInt(recordHeader.otherType);
      dataOutputStream.writeVLong(recordHeader.otherCrc32);
      dataOutputStream.flush();
      checkedOutputStream.flush();
      CRC32 crc32 = (CRC32) checkedOutputStream.getChecksum();
      long crcValue = crc32.getValue();

      int headersSize = byteOutput.size();

      OutputStream fileOutputStream = output.getOutputStream(currentWritePosition);
      BufferedOutputStream bufferedFileOutput = new BufferedOutputStream(fileOutputStream);
      CountingOutputStream countingOutput = new CountingOutputStream(bufferedFileOutput);
      VDataOutputStream vOutput = new VDataOutputStream(countingOutput);
      vOutput.writeVLong(crcValue);
      vOutput.writeVInt(headersSize);
      int sizeOfInitial = countingOutput.getCount();
      recordHeader.headersLength = headersSize + sizeOfInitial;

      byteOutput.writeTo(vOutput);
      if (docBytes != null)
        docBytes.writeTo(vOutput);
      if (otherBytes != null)
        otherBytes.writeTo(vOutput);
      vOutput.flush();
      bufferedFileOutput.flush();
      fileOutputStream.flush();
      /**
       * output.write(HEADER); output.writeByte(OK); output.writeLong(id);
       * output.writeInt(recordHeader.docsLength);
       * output.writeInt(recordHeader.otherLength);
       */
      /**
       * recordHeader.docsPosition = output.getFilePointer(); if (docBytes !=
       * null) output.write(docBytes); if (otherBytes != null)
       * output.write(otherBytes); int totalLength = recordHeader.docsLength +
       * recordHeader.otherLength+3+1+8+4+4; System.out.println("write record
       * length: "+totalLength);
       */
      output.getFD().sync();
      LOG.info("currentWritePosition: " + currentWritePosition+" length: "+recordHeader.getLength()+" docs length: "+recordHeader.docsLength);
      currentWritePosition = output.getFilePointer();
      return recordHeader;
    } finally {
      writeLock.unlock();
    }
  }

  public void close() throws IOException {
    output.close();
  }
}
